package com.iocup.keybastion.spring.starter.store;

import com.iocup.keybastion.core.store.AbstractSessionStore;
import com.iocup.keybastion.exception.SessionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.ConvertingCursor;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.serializer.RedisSerializer;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author xyjxust
 * @create 2019/10/23 16:07
 **/
@Slf4j
public class RedisCacheStore extends AbstractSessionStore {

    private RedisTemplate redisTemplate;

    public RedisCacheStore(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }


    @Override
    public void setValue(String key, Object value, long expire) {
        if (StringUtils.isBlank(key) || value == null) {
            throw new SessionException("key或者value不能为空");
        }
        if (expire == NO_EXPIRE) {
            this.redisTemplate.opsForValue().set(key, value);
            return;
        }
        this.redisTemplate.opsForValue().set(key, value, expire, TimeUnit.SECONDS);
    }

    @Override
    public void updateValue(String key, Object value) {
        long expireTime = getTimeout(key);
        if (expireTime == NOT_KEY) {
            return;
        }
        if (expireTime == NO_EXPIRE) {
            this.redisTemplate.opsForValue().set(key, value);
        }
        this.redisTemplate.opsForValue().set(key, value, expireTime, TimeUnit.SECONDS);
    }

    @Override
    public void setExpire(String key, long expire) {
        this.redisTemplate.expire(key, expire, TimeUnit.SECONDS);
    }

    @Override
    public <T> T getValue(String key) {
        Object value = this.redisTemplate.opsForValue().get(key);
        if (value == null) {
            return null;
        }
        return (T) value;
    }

    @Override
    public void removeValue(String key) {
        this.redisTemplate.delete(key);
    }

    @Override
    public long getTimeout(String key) {
        if (!this.redisTemplate.hasKey(key)) {
            return NOT_KEY;
        }
        Long expire = this.redisTemplate.getExpire(key);
        if (expire == null || expire == 0) {
            return NOT_KEY;
        }
        return expire;
    }

    @Override
    public List<String> queryPrefixKey(String prefKey, int start, int size) {
        return searchList(prefixKeys(prefKey), prefKey, start, size);
    }


    public Set<String> prefixKeys(String prefKey) {
        Set<String> keys = new HashSet<>();
        Cursor<String> cursor = null;
        try {
            cursor = this.scan(redisTemplate, prefKey + "*", 100L);
            while (cursor.hasNext()) {
                keys.add(String.valueOf(cursor.next()));
            }
        } catch (Exception e) {
            log.error("get keys error: ", e);
        } finally {
            if (cursor != null) {
                try {
                    cursor.close();
                } catch (IOException e) {
                    log.error("close cursor error", e);
                }
            }
        }

        return keys;
    }

    /**
     * scan 实现  使用scan查询所有key 避免使用keys命令引发Redis锁
     *
     * @param pattern 表达式
     * @param limit   每次迭代的条数
     */
    public Cursor<String> scan(RedisTemplate<String, ?> redisTemplate, String pattern, Long limit) {
        ScanOptions options = new ScanOptions.ScanOptionsBuilder().match(pattern).count(limit).build();
        RedisSerializer<String> redisSerializer = (RedisSerializer<String>) redisTemplate.getKeySerializer();
        AtomicReference<RedisConnection> redisConnection1 = new AtomicReference<>();
        Cursor<String> cursor = redisTemplate.executeWithStickyConnection(redisConnection -> {
            redisConnection1.set(redisConnection);
            return new ConvertingCursor<>(redisConnection.scan(options), redisSerializer::deserialize);
        });
        return cursor;
    }
}
